package com.study.spring.hadoop;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.filter.FilterBase;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class RowKeyFilter extends FilterBase {

    public Map<Object, Object> map = new HashMap<Object, Object>();
    /**
     * 过车开始时间
     */
    private long start;
    /**
     * 过车结束时间
     */
    private long end;
    private byte[] value;
    private boolean filterRow = true;

    public RowKeyFilter() {
        super();
    }

    public RowKeyFilter(byte[] value) {
        this.value = value;
    }

    public RowKeyFilter(long start, long end) {
        this.start = start;
        this.end = end;
    }


    public ReturnCode filterKeyValue(KeyValue ignored) {
        if (!this.filterRow)
            return ReturnCode.INCLUDE;
        else
            return ReturnCode.NEXT_ROW;
    }

    @Override
    public boolean filterRowKey(Cell cell) throws IOException {
        byte[] bytes = cell.getRowArray();
        System.out.println(new String(bytes));
        return super.filterRowKey(cell);
    }

    @Override
    public boolean filterRow() throws IOException {
        return true;
    }

    @Override
    public void reset() throws IOException {
        this.filterRow = true;
    }


    @Override
    public boolean filterRowKey(byte[] buffer, int offset, int length) throws IOException {
        byte[] rowKey = Arrays.copyOfRange(buffer, offset, offset + length);
        String str = new String(rowKey);
        if (map.containsKey(str)) { // 在0(1)时间内返回，效率较高
            this.filterRow = false; // false表示包括这一行
        }
        return this.filterRow;
    }

    @Override
    public boolean filterAllRemaining() throws IOException {
        return super.filterAllRemaining();
    }

    @Override
    public Cell transformCell(Cell v) throws IOException {
        return super.transformCell(v);
    }

    @Override
    public void filterRowCells(List<Cell> ignored) throws IOException {
        super.filterRowCells(ignored);
    }

    @Override
    public boolean hasFilterRow() {
        return super.hasFilterRow();
    }

    @Override
    public Cell getNextCellHint(Cell currentCell) throws IOException {
        return super.getNextCellHint(currentCell);
    }

    @Override
    public boolean isFamilyEssential(byte[] name) throws IOException {
        return super.isFamilyEssential(name);
    }

    @Override
    public String toString() {
        return super.toString();
    }

    @Override
    public byte[] toByteArray() throws IOException {
        return super.toByteArray();
    }

    @Override
    public ReturnCode filterKeyValue(Cell c) throws IOException {
        return super.filterKeyValue(c);
    }

    @Override
    public ReturnCode filterCell(Cell c) throws IOException {
        return super.filterCell(c);
    }

    @Override
    public boolean isReversed() {
        return super.isReversed();
    }

    @Override
    public void setReversed(boolean reversed) {
        super.setReversed(reversed);
    }


}
